package com.atguigu.edu.app.dwd.db;

import com.atguigu.edu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradePaySuc {
    public static void main(String[] args) {
        //TODO 1 环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO　2 设置状态后端
                 /*
                 env.enableCheckpointing(5 * 60 * 1000L, CheckpointingMode.EXACTLY_ONCE );
                 env.getCheckpointConfig().setCheckpointTimeout( 3 * 60 * 1000L );
                 env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
                 env.setStateBackend(new HashMapStateBackend());
                 env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
                 System.setProperty("HADOOP_USER_NAME", "atguigu");
                  */
        //TODO 3 读取kafka对应主题topic_db的数据
        String topicName = "topic_db";
        String groupID = "dwd_trade_pay_suc";
        tableEnv.executeSql("CREATE TABLE topic_db (\n" +
                "  `database` string,\n" +
                "  `table` string,\n" +
                "  `type` string,\n" +
                "  `ts` bigint,\n" +
                "  `xid` bigint,\n" +
                "  `commit` string,\n" +
                "  `data` map<string,string>,\n" +
                "  `old` map<string,string> \n" +
//                "   pt AS PROCTIME() \n" +
                ") " + KafkaUtil.getKafkaDDL(topicName, groupID));

        //TODO 4 过滤出下单成功信息
        Table paySucTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['alipay_trade_no'] alipay_trade_no,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['payment_type'] payment_type,\n" +
                "`data`['payment_status'] payment_status,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`data`['callback_content'] callback_content,\n" +
                "`data`['callback_time'] callback_time\n" +
                "from topic_db\n" +
                "where `table` ='payment_info'\n" +
                "and `type` ='insert' \n" +
                "and `data`['payment_status'] ='1602'");
        tableEnv.createTemporaryView("payment_info",paySucTable);


        //TODO 5 过滤出订单表
        Table orderInfoTable = tableEnv.sqlQuery("select \n" +
                "  `data`['id'] id,\n" +
                "  `data`['order_status'] order_status,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['origin_amount'] origin_amount,\n" +
                "  `data`['coupon_reduce'] coupon_reduce,\n" +
                "  `data`['final_amount'] final_amount,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `data`['province_id'] province_id,\n" +
                "  `data`['session_id'] session_id,\n" +
                "  `ts`\n" +
                "from topic_db\n" +
                "where`table`='order_info'\n" +
                "and `type`='insert' ");
        tableEnv.createTemporaryView("order_info",orderInfoTable);


        //TODO 6 关联订单表和下单表
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "pi.id,\n" +
                "oi.user_id,\n" +
                "pi.out_trade_no,\n" +
                "pi.order_id,\n" +
                "pi.alipay_trade_no,\n" +
                "pi.total_amount,\n" +
                "pi.trade_body,\n" +
                "pi.payment_type,\n" +
                "pi.payment_status,\n" +
                "pi.create_time,\n" +
                "pi.update_time,\n" +
                "pi.callback_content,\n" +
                "pi.callback_time\n" +
                "from order_info oi\n" +
                "join payment_info pi\n" +
                "on oi.id= pi.order_id");
        tableEnv.createTemporaryView("result_table",resultTable);


        //TODO 5 写出到kafka新的主题
        tableEnv.executeSql("create table pay_success(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "out_trade_no string,\n" +
                "order_id string,\n" +
                "alipay_trade_no string,\n" +
                "total_amount string,\n" +
                "trade_body string,\n" +
                "payment_type string,\n" +
                "payment_status string,\n" +
                "create_time string,\n" +
                "update_time string,\n" +
                "callback_content string,\n" +
                "callback_time string\n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_trade_pay_detail_suc"));
        tableEnv.executeSql("insert into pay_success select * from result_table");

    }
}
